package com.lanou.quanttradedata.stock.mapreduce.map;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 映射类：
 * 用来将某只股票，所有HDFS上保存的日线数据进行映射和筛选
 * 最终筛选出近7天的数据，进入reduce阶段
 */
public class StockDailyInfoAvgMap  extends Mapper<LongWritable, Text, Text, Text> {
    private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        if(key.get() != 0) {
            //只要不是表头行，就是数据行，数据行进行map筛选部分
            //数据行样例: P"689009.SH","20230919",34.23,34.45,33.7,34.24,34.2,0.04,0.117,21591.9,73595.898
            String[] line = value.toString().split(",");
            //股票代码
            String lineKey = line[0].substring(2);
            //近7天
            long d = 1000L*60*60*24*7;
            //获得7天前的日期
            Integer start = Integer.valueOf(sdf.format(new Date(System.currentTimeMillis() - d)));
            //获取当前这行日线数据 是哪天的
            Integer temp = Integer.valueOf(line[1].replace("\"", ""));
            //如果日线数据是近7天的
            if(temp >= start) {
                //将数据写入reduce
                context.write(new Text(lineKey), value);
            }
        }
    }
}
